package com.idanchuang.component.mq.amqp.switcher.apollo;

import com.ctrip.framework.apollo.model.ConfigChange;
import com.ctrip.framework.apollo.model.ConfigChangeEvent;
import com.ctrip.framework.apollo.spring.annotation.ApolloConfigChangeListener;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author yjy
 * Created at 2021/8/9 1:28 下午
 */
public class ApolloConsumeSwitcher {

    private static final Logger logger = LoggerFactory.getLogger(ApolloConsumeSwitcher.class);
    /**
     * apollo key: 值是队列名, 多个队列名用英文逗号分割
     * 如: idanchuang.component.mq.stopQueues = my-queue1,my-queue2
     */
    public static final String STOP_CONSUME_KEY = "idanchuang.component.mq.stopQueues";

    /**
     * 所有的队列监听容器MAP
     */
    private static final Map<String, AbstractMessageListenerContainer> ALL_QUEUE2_CONTAINER_MAP = new ConcurrentHashMap<>();

    public void init(Collection<MessageListenerContainer> containers, String initConfig) {
        // init container map
        containers.forEach(container -> {
            AbstractMessageListenerContainer simpleContainer = (AbstractMessageListenerContainer) container;
            Arrays.stream(simpleContainer.getQueueNames()).forEach(queueName ->
                    ALL_QUEUE2_CONTAINER_MAP.putIfAbsent(StringUtils.trim(queueName), simpleContainer));
        });
        if (initConfig != null && !initConfig.isEmpty()) {
            stopQueue(initConfig);
        }
        logger.info("ApolloConsumeSwitcher init > container size: {}", ALL_QUEUE2_CONTAINER_MAP.size());
    }

    @ApolloConfigChangeListener(value = "application", interestedKeys = {
            STOP_CONSUME_KEY})
    public void doChange(ConfigChangeEvent event) {
        if (ALL_QUEUE2_CONTAINER_MAP.isEmpty()) {
            return;
        }
        ConfigChange queues = event.getChange(STOP_CONSUME_KEY);
        String oldValue = queues.getOldValue();
        String newValue = queues.getNewValue();
        String[] startQueues, stopQueues;
        switch (queues.getChangeType()) {
            case ADDED:
                stopQueues = newValue.split(",");
                stopQueue(stopQueues);
                break;
            case MODIFIED:
                if (newValue == null || newValue.isEmpty()) {
                    startQueues = oldValue.split(",");
                    startQueue(startQueues);
                    break;
                }
                if (oldValue == null || oldValue.isEmpty()) {
                    stopQueues = newValue.split(",");
                    stopQueue(stopQueues);
                    break;
                }
                String[] oldValues = oldValue.split(",");
                String[] newValues = newValue.split(",");
                Set<String> oldSet = new HashSet<>(Arrays.asList(oldValues));
                Set<String> newSet = new HashSet<>(Arrays.asList(newValues));
                Set<String> retainSet = new HashSet<>(oldSet);
                retainSet.retainAll(newSet);
                // restart queues
                oldSet.removeAll(retainSet);
                for (String queue : oldSet) {
                    startQueue(queue);
                }
                // stop queues
                newSet.removeAll(retainSet);
                for (String queue : newSet) {
                    stopQueue(queue);
                }
                break;
            case DELETED:
                startQueues = oldValue.split(",");
                startQueue(startQueues);
                break;
            default:
                break;
        }
    }

    /**
     * 启用队列
     * @param queues 队列名
     */
    private void startQueue(String... queues) {
        if (queues == null || queues.length == 0) {
            return;
        }
        for (String queue : queues) {
            if (queue.isEmpty()) {
                continue;
            }
            AbstractMessageListenerContainer container = ALL_QUEUE2_CONTAINER_MAP.get(queue);
            if (container == null) {
                logger.info("MessageListenerContainer start cancel, cause container is null > queue: {}", queue);
                continue;
            }
            container.start();
            logger.info("MessageListenerContainer start > listenerId: {}, queueName: {}",
                    container.getListenerId(), queue);
        }
    }

    /**
     * 停用队列
     * @param queues 队列名
     */
    private void stopQueue(String... queues) {
        if (queues == null || queues.length == 0) {
            return;
        }
        for (String queue : queues) {
            if (queue.isEmpty()) {
                continue;
            }
            AbstractMessageListenerContainer container = ALL_QUEUE2_CONTAINER_MAP.get(queue);
            if (container == null) {
                logger.info("MessageListenerContainer stop cancel, cause container is null > queue: {}", queue);
                continue;
            }
            container.stop(() -> {
                logger.info("MessageListenerContainer stop > listenerId: {}, queueName: {}",
                        container.getListenerId(), queue);
            });
        }
    }

    public static Map<String, AbstractMessageListenerContainer> getAllQueue2ContainerMap() {
        return new HashMap<>(ALL_QUEUE2_CONTAINER_MAP);
    }


}
